package org.apache.rocketmq.store.queue;


import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.message.MessageExtBrokerInner;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.store.DefaultMessageStore;
import org.apache.rocketmq.store.DispatchRequest;
import org.apache.rocketmq.store.config.MessageStoreConfig;
import org.rocksdb.RocksDBException;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;


/**
 * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
 * @author：陈清风扬，个人微信号：chenqingfengyangjj。
 * @date:2025/6/29
 * @方法描述：消费队列管理器的抽象父类，在第十四版本代码中，这个类的内容还很简单
 */
public abstract class AbstractConsumeQueueStore implements ConsumeQueueStoreInterface {

    protected static final Logger log = LoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);

    protected final DefaultMessageStore messageStore;

    protected final MessageStoreConfig messageStoreConfig;

    //消费队列偏移量管理器
    protected final QueueOffsetOperator queueOffsetOperator = new QueueOffsetOperator();

    //消费队列集合，每一个主题都对应这个一个消费队列集合
    //key就是主题名称，value就是该主题下的所有消息队列，这个value的key就是消费队列Id，value就是对应的ConsumeQueue消费队列对象
    protected final ConcurrentMap<String/* topic */, ConcurrentMap<Integer/* queueId */, ConsumeQueueInterface>> consumeQueueTable;


    public AbstractConsumeQueueStore(DefaultMessageStore messageStore) {
        this.messageStore = messageStore;
        this.messageStoreConfig = messageStore.getMessageStoreConfig();
        this.consumeQueueTable = new ConcurrentHashMap<>(32);
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：为要存储的消息分配消息队列偏移量的方法，这个方法就会把本条消息在ConsumeQueue目标队列中的偏移量明确了
     */
    @Override
    public void assignQueueOffset(MessageExtBrokerInner msg) throws RocksDBException {
        //根据主题和队列Id找到对应的消费队列ConsumeQueue对象
        ConsumeQueueInterface consumeQueue = findOrCreateConsumeQueue(msg.getTopic(), msg.getQueueId());
        consumeQueue.assignQueueOffset(this.queueOffsetOperator, msg);
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：增加ConsumeQueue消费队列偏移量的方法，其实就是记录一下这个消息队列中又存储了多少条消息
     */
    @Override
    public void increaseQueueOffset(MessageExtBrokerInner msg, short messageNum) {
        ConsumeQueueInterface consumeQueue = findOrCreateConsumeQueue(msg.getTopic(), msg.getQueueId());
        consumeQueue.increaseQueueOffset(this.queueOffsetOperator, msg, messageNum);
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：消费队列存储器存放被分发的消息的方法
     */
    @Override
    public void putMessagePositionInfoWrapper(ConsumeQueueInterface consumeQueue, DispatchRequest request) {
        consumeQueue.putMessagePositionInfoWrapper(request);
    }

    @Override
    public Long getMaxOffset(String topic, int queueId) {
        return this.queueOffsetOperator.currentQueueOffset(topic + "-" + queueId);
    }

    @Override
    public void setTopicQueueTable(ConcurrentMap<String, Long> topicQueueTable) {
        this.queueOffsetOperator.setTopicQueueTable(topicQueueTable);
        this.queueOffsetOperator.setLmqTopicQueueTable(topicQueueTable);
    }

    @Override
    public ConcurrentMap getTopicQueueTable() {
        return this.queueOffsetOperator.getTopicQueueTable();
    }

    @Override
    public void increaseLmqOffset(String queueKey, short messageNum) {
        queueOffsetOperator.increaseLmqOffset(queueKey, messageNum);
    }

    @Override
    public long getLmqQueueOffset(String queueKey) {
        return queueOffsetOperator.getLmqOffset(queueKey);
    }

    @Override
    public void removeTopicQueueTable(String topic, Integer queueId) {
        this.queueOffsetOperator.remove(topic, queueId);
    }

    @Override
    public ConcurrentMap<String, ConcurrentMap<Integer, ConsumeQueueInterface>> getConsumeQueueTable() {
        return this.consumeQueueTable;
    }

    @Override
    public ConcurrentMap<Integer, ConsumeQueueInterface> findConsumeQueueMap(String topic) {
        return this.consumeQueueTable.get(topic);
    }

    // @Override
    // public long getStoreTime(CqUnit cqUnit) {
    //     if (cqUnit != null) {
    //         try {
    //             final long phyOffset = cqUnit.getPos();
    //             final int size = cqUnit.getSize();
    //             long storeTime = this.messageStore.getCommitLog().pickupStoreTimestamp(phyOffset, size);
    //             return storeTime;
    //         } catch (Exception e) {
    //         }
    //     }
    //     return -1;
    // }
}
